package com.atguigu.gmall.realtime.util;

import com.atguigu.gmall.realtime.common.Constant;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * @Author lzc
 * @Date 2022/12/26 14:59
 */
public class FlinkSourceUtil {
    
    public static KafkaSource<String> getKafkaSource(String groupId, String topic) {
        // 隔离级别: 默认是消费所有数据(提交和未提交)
        // "isolation.level" => default:read_uncommitted(有可能重复消费)   read_committed
    
        return KafkaSource.<String>builder()
            .setBootstrapServers(Constant.KAFKA_BROKERS)
            .setGroupId(groupId)
            .setTopics(topic)  // ctrl+shift+u
            .setStartingOffsets(OffsetsInitializer.latest())  // 默认从什么位置开始消费.如果设置了 checkpoint 则从 checkpoint 的最后一次正确的位置消费.如果第一次,则从这里的设置的位置开始消费
            // 当 left join 的时候, 会向 kafka 写入 null, SimpleStringSchema会直接反序列化 null 值, 会导致空指针异常
//            .setValueOnlyDeserializer(new SimpleStringSchema())
            .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                // 返序列化
                @Override
                public String deserialize(byte[] message) throws IOException {
                    // 需要把字节数组变成字符串. 如果写入的是 null,则这里的message也是 null
                    if (message != null) {
                        return new String(message, StandardCharsets.UTF_8);
                    }
                    return null;
                }
    
                // 要不要结束流
                // 如果是无界流, 永远返回 false
                @Override
                public boolean isEndOfStream(String nextElement) {
                    return false;
                }
    
                @Override
                public TypeInformation<String> getProducedType() {
//                    return Types.STRING;  // ①特殊类型: 基本类型 string 元组
//                    return TypeInformation.of(String.class); // ②特殊类型: 基本类型 string 元组, 普通的类型
                    return TypeInformation.of(new TypeHint<String>() {}); // ③最通用, 所有都可以用, 比如类型中包含泛型
                    
                    // 使用范围: ③>②>①
                }
            })
            .setProperty("isolation.level", "read_committed")
            .build();
    }
}
